package com.hsc.www.kafka;


import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class Producer {

    private KafkaProducer<String, String> producer;


    private Producer() {
        Properties props = new Properties();
        //kafka集群地址
        props.put("bootstrap.servers", "172.172.172.151:9092");
        //应答方式,0:不等待kafka服务器的ack响应，容易出现数据丢失，retries 尝试参数将会失效。1、
        props.put("acks", "all");//
        props.put("retries", 0);//retries = MAX 无限重试，直到你意识到出现了问题:)
        props.put("batch.size", 16384);//producer将试图批处理消息记录，以减少请求次数.默认的批量处理消息字节数
        //batch.size当批量的数据大小达到设定值后，就会立即发送，不顾下面的linger.ms
        props.put("linger.ms", 1);//延迟1ms发送，这项设置将通过增加小的延迟来完成--即，不是立即发送一条记录，producer将会等待给定的延迟时间以允许其他消息记录发送，这些消息记录可以批量处理
        props.put("buffer.memory", 33554432);//producer可以用来缓存数据的内存大小。
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        producer = new KafkaProducer<>(props);
    }

    public void sendMessage() {
        String topic = "111111";
        for (int i = 1; i <= 500; i++) {
            producer.send(new ProducerRecord<>(topic, i + "", "消息topic=" + topic + "no=" + i));
        }
        producer.close();
    }

    public static void main(String[] args) throws InterruptedException {
        new Producer().sendMessage();
        Thread.sleep(1000);
    }
}